package org.latestbit.smartobjects.perftests.client;

import java.util.concurrent.atomic.AtomicInteger;

import org.latestbit.smartobjects.ISmartBus;
import org.latestbit.smartobjects.exceptions.*;
import org.latestbit.smartobjects.perftests.*;

public class AsyncPerfClientImpl extends AbstractPerfTestImpl implements Runnable, IPerfTestServiceCallback {
	AtomicInteger tpsSent = new AtomicInteger(0);
	AtomicInteger tpsRcvd = new AtomicInteger(0);
	private Thread statThread;

	public AsyncPerfClientImpl(
			ISmartBus bus, 
			CommandLineOptions options,
			IPerfTestService service) {
		super(bus, options, service);
		this.statThread = new Thread(new StatThread());		
	}

	@Override
	public void start() throws Exception {
		statThread.start();
		for(int i=0; i< options.getThreadsCount(); i++)
			threadPool.execute(this);		
	}

	@Override
	public void stop() {
		stopFlag = true;
		try {
			statThread.join();
		} 
		catch (Exception e) {
			e.printStackTrace();
		}
		threadPool.shutdown();
	}

	@Override
	public void run() {
		while(!stopFlag) {
			try {
				this.service.testMethodRequest(
						createTestParamStr(), 
						createTestParamInt(), 
						createTestStructure(), 
						createTestStructureArray(this.options.getArraySize()),
						this
				);
				tpsSent.addAndGet(1);
				if(options.getDelay()!=null) {
					Thread.sleep(options.getDelay());
				}				
			} 
			catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}
	
	protected class StatThread implements Runnable {

		@Override
		public void run() {
			while(!stopFlag) {
				try {
					Thread.sleep(1000);
					int lTpsSent = tpsSent.getAndSet(0);
					int lTpsRcvd = tpsRcvd.getAndSet(0);
					long bytesSnt = lTpsSent * packetSize;
					long bytesRcvd = lTpsRcvd * packetSize;
					long kbitSnt = (bytesSnt/1024) * 8;
					long kbitRcvd = (bytesRcvd/1024)*8;
					long mbitSnt = (bytesSnt/1024/1024)*8;
					long mbitRcvd = (bytesRcvd/1024/1024)*8;
					
					System.out.println(
							"TPS Sent: "+ lTpsSent+" / TPS Rcvd: "+lTpsRcvd
							+" : "+bytesSnt+" bytes / "+bytesRcvd +" bytes "
							+( mbitSnt <= 1 ?
									" ( "+kbitSnt+" Kbit / "+kbitRcvd+" Kbit )"
									: 
									" ( "+mbitSnt+" Mbit / "+mbitRcvd+" Mbit )"								
							)
					);
				} 
				catch (Exception e) {
					e.printStackTrace();
				}				
			}
		}
		
	}

	@Override
	public void onResponseTimeout(String operationId, Object userObject) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void onResponseSystemException(String operationId,
			SmartBusSystemException ex, Object userObject) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void onTestMethodResponse(ITestStructureType value, Object userObject) {
		tpsRcvd.addAndGet(1);
	}

	
}
